// Copyright (C) 2012 Yury Bychenkov (http://bychenkov.com).
// All rights reserved.
//
// Redistribution and use in source and binary forms, with or without
// modification, are permitted provided that the following conditions are
// met:
//
//  - Redistributions of source code must retain the above copyright
// notice, this list of conditions and the following disclaimer.
//  - Redistributions in binary form must reproduce the above
// copyright notice, this list of conditions and the following disclaimer
// in the documentation and/or other materials provided with the
// distribution.
//  - Neither the name of Yury Bychenkov nor the names of its
// contributors may be used to endorse or promote products derived from
// this software without specific prior written permission.
//
// THIS SOFTWARE IS PROVIDED BY THE COPYRIGHT HOLDERS AND CONTRIBUTORS
// "AS IS" AND ANY EXPRESS OR IMPLIED WARRANTIES, INCLUDING, BUT NOT
// LIMITED TO, THE IMPLIED WARRANTIES OF MERCHANTABILITY AND FITNESS FOR
// A PARTICULAR PURPOSE ARE DISCLAIMED. IN NO EVENT SHALL THE COPYRIGHT
// OWNER OR CONTRIBUTORS BE LIABLE FOR ANY DIRECT, INDIRECT, INCIDENTAL,
// SPECIAL, EXEMPLARY, OR CONSEQUENTIAL DAMAGES (INCLUDING, BUT NOT
// LIMITED TO, PROCUREMENT OF SUBSTITUTE GOODS OR SERVICES; LOSS OF USE,
// DATA, OR PROFITS; OR BUSINESS INTERRUPTION) HOWEVER CAUSED AND ON ANY
// THEORY OF LIABILITY, WHETHER IN CONTRACT, STRICT LIABILITY, OR TORT
// (INCLUDING NEGLIGENCE OR OTHERWISE) ARISING IN ANY WAY OUT OF THE USE
// OF THIS SOFTWARE, EVEN IF ADVISED OF THE POSSIBILITY OF SUCH DAMAGE.

template <typename Consumer>
Database::ProxyConsumer<Consumer>::ProxyConsumer(Consumer& consumer_, uint topCount_):
	consumer(consumer_),
	topCount(topCount_)
{
}

template <typename Consumer>
bool Database::ProxyConsumer<Consumer>::isDone() const {
	return topCount == 0;
}

template <typename Consumer>
bool Database::ProxyConsumer<Consumer>::put(const uint8* tupleData) {
	if (topCount == 0) {
		return false;
	}
	topCount--;
	return consumer.put(tupleData) && topCount > 0;
}

inline Database::OrderingConsumer::OrderingConsumer(uint8* memory, uint memorySize, uint topCount):
	tupleHeap(memory, memorySize, topCount)
{
}

inline bool Database::OrderingConsumer::put(const uint8* tupleData) {
	return tupleHeap.put(tupleData);
}

template <typename Consumer> 
bool Database::search(const char* queryString, Consumer& consumer, uint& count) {
	count = 0;

	Query* query = QueryCompiler::compile(schema, queryString);
	if (query == 0) {
		return false;
	}
	
	
	consumer.begin(schema, query->getTupleColumns());

	uint topCount = query->getTopCount();

	ProxyConsumer<Consumer> proxyConsumer(consumer, topCount == 0 ? 0xFFFFFFFFUL : topCount);

	const bool hasOrdering = query->hasOrdering();

	uint8* orderingHeapMemory = 0;
	uint orderingHeapMemorySize = 0;


	if (hasOrdering) {	
		orderingHeapMemorySize = 256 * 1024 * 1024; // TODO config
		orderingHeapMemory = new (nothrow) uint8[orderingHeapMemorySize];
		if (orderingHeapMemory == 0) {
			logger.error("Can't allocate memory for ordering");
			delete query;
			return false;
		}
   	}
	
	OrderingConsumer orderingConsumer(orderingHeapMemory, orderingHeapMemorySize / 2, topCount);

	bool success = true;

	Chunk chunk;
	for (uint i = 0; !proxyConsumer.isDone() && chunks.get(i, chunk); i++) {
		if (chunk.nunChunk != 0) {
			NunChunkProvider provider(schema, *chunk.nunChunk);
			QueryExecutor<NunChunkProvider>* executor = QueryExecutor<NunChunkProvider>::link(*query, provider);
			if (executor == 0) {
			    success = false;
				break;
			}
			if (hasOrdering) {
				count += executor->execute(orderingConsumer, orderingHeapMemory + orderingHeapMemorySize / 2, orderingHeapMemorySize / 2);
			} else {
				count += executor->execute(proxyConsumer, 0, 0);
			}
			delete executor;
		} else {
			LogChunkProvider provider(schema, *chunk.logChunk);
			QueryExecutor<LogChunkProvider>* executor = QueryExecutor<LogChunkProvider>::link(*query, provider);
			if (executor == 0) {
			    success = false;
				break;
			}
			if (hasOrdering) {
				count += executor->execute(orderingConsumer, orderingHeapMemory + orderingHeapMemorySize / 2, orderingHeapMemorySize / 2);
			} else {
				count += executor->execute(proxyConsumer, 0, 0);
			}
			delete executor;
		}
	}

	if (success && hasOrdering) {
		for (uint i = 0, k = orderingConsumer.tupleHeap.getCount(); i < k; i++) {
			if (!proxyConsumer.put(orderingConsumer.tupleHeap.peek())) {
				break;
			}
			orderingConsumer.tupleHeap.remove();
		}
	}

	delete query;
	if (orderingHeapMemory != 0) {
		delete [] orderingHeapMemory;
	}

	return success;
}
